package com.haidechizi.longpolling.client.service;

import com.haidechizi.longpolling.client.entity.LongPollingDto;
import com.haidechizi.longpolling.client.entity.Response;
import com.haidechizi.longpolling.constant.CommonConstant;
import com.haidechizi.longpolling.event.Event;
import com.haidechizi.longpolling.event.LocalDataChangeEvent;
import com.haidechizi.longpolling.event.listener.SingleEventListener;
import com.haidechizi.longpolling.server.entity.ConfigData;
import com.haidechizi.longpolling.server.persistent.ConfigDataPersistent;
import com.haidechizi.longpolling.utils.ResponseUtils;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.servlet.AsyncContext;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author wangdejun
 * @create 2021/2/24 11:20
 */
@Service
public class LongPollingService extends SingleEventListener {

    @Value("${pool.with.fix.rate:false}")
    private boolean isFixPoll;

    private static final ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 8,
            60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100), new ThreadPoolExecutor.CallerRunsPolicy());

    private static final ScheduledExecutorService executorService = Executors.newScheduledThreadPool(4);

    private static final Queue<LongPollingClient> longPollingClientQueue = new ConcurrentLinkedQueue<>();

    public void addLongPollingClient(LongPollingDto longPollingDto, HttpServletRequest request, HttpServletResponse response) {
        String timeoutStr = request.getHeader(CommonConstant.LONG_POLLING_TIMEOUT);
        int timeout = Math.max(10000, Integer.valueOf(timeoutStr) - 500);
        if (isFixPolling()) {
            timeout = 10000;
        } else {
            // 检测数据是否发生变化
            ConfigData configData = ConfigDataPersistent.getConfigData(longPollingDto.getAppCode());
            if (configData != null && !configData.getMd5().equals(longPollingDto.getMd5())) {
                // 数据发生变化，直接返回
                ResponseUtils.generatorResponse(response, Response.builder().data(configData));
                return;
            }
        }
        // 开启异步
        AsyncContext asyncContext = request.startAsync();

        // 通过线程池异步请求
        executor.execute(new LongPollingClient(longPollingDto.getAppCode(), longPollingDto.getMd5(), asyncContext, timeout));

    }

    private boolean isFixPolling() {
        return isFixPoll;
    }

    @Override
    public Class<? extends Event> interest() {
        return LocalDataChangeEvent.class;
    }

    /**
     * 触发事件
     *
     * @param event
     */
    @Override
    public void onEvent(Event event) {
        if (isFixPolling()) {
            // 忽略，待定时执行时返回变更的数据
        } else {
            if (event instanceof LocalDataChangeEvent) {
                LocalDataChangeEvent localDataChangeEvent = (LocalDataChangeEvent) event;
                executor.execute(new DataChangeTask(localDataChangeEvent.getAppCode()));
            }
        }
    }

    @Data
    class LongPollingClient implements Runnable {

        private final String appCode;
        private final String md5;
        private final AsyncContext asyncContext;
        private final int timeout;

        Future<?> asyncTimeoutFuture;

        public LongPollingClient(String appCode, String md5, AsyncContext asyncContext, int timeout) {
            this.appCode = appCode;
            this.md5 = md5;
            this.asyncContext = asyncContext;
            this.timeout = timeout;
        }

        @Override
        public void run() {
            asyncTimeoutFuture = executorService.schedule(() -> {
                longPollingClientQueue.remove(this);
                // 固定时间的模式通过此定时返回变更的数据
                if (isFixPolling()) {
                    ConfigData configData = ConfigDataPersistent.getConfigData(appCode);
                    if (configData != null && !md5.equals(configData.getMd5())) {
                        sendResponse(configData);
                    } else {
                        sendResponse(null);
                    }
                } else {
                    // 无数据变更，只是使用此方式hold请求，
                    sendResponse(null);
                }

            }, timeout, TimeUnit.MILLISECONDS);
            longPollingClientQueue.add(this);

        }

        void sendResponse(ConfigData configData) {
            // 取消定时
            if (asyncTimeoutFuture != null) {
                asyncTimeoutFuture.cancel(false);
            }
            generatorResponse(configData);
        }

        void generatorResponse(ConfigData configData) {
            if (configData != null) {
                ResponseUtils.generatorResponse((HttpServletResponse) asyncContext.getResponse(), Response.builder().data(configData));
            }
            // 结束hold住的请求
            asyncContext.complete();
        }
    }

    class DataChangeTask implements Runnable {

        private final String appCode;

        public DataChangeTask(String appCode) {
            this.appCode = appCode;
        }

        @Override
        public void run() {
            if (CollectionUtils.isEmpty(longPollingClientQueue)) {
                return;
            }
            Iterator<LongPollingClient> iterator = longPollingClientQueue.iterator();
            while (iterator.hasNext()) {
                LongPollingClient longPollingClient = iterator.next();
                if (appCode.equals(longPollingClient.getAppCode())) {
                    // 移除持有的请求
                    iterator.remove();
                    ConfigData configData = ConfigDataPersistent.getConfigData(appCode);
                    longPollingClient.sendResponse(configData);
                }

            }
        }
    }
}
